package kafka;

import com.vcg.kafka.KafkaConsumerApplication;
import lombok.extern.java.Log;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.*;
import java.util.stream.Collectors;

/**
 * Created by wuyu on 2016/9/20.
 */
@SpringBootTest(classes = KafkaConsumerApplication.class)
@RunWith(SpringRunner.class)
@Log
public class HelloWorld {

    @Autowired
    ConsumerFactory<Integer, String> consumerFactory;

    @Test
    public void testRec() throws InterruptedException {
        Consumer<Integer, String> consumer = consumerFactory.createConsumer();
        Map<String, List<PartitionInfo>> stringListMap = consumer.listTopics();
        log.info(stringListMap.toString());
       consumer.subscribe(Arrays.asList("test"));
        while(true){
            Thread.sleep(100);
//            log.info("订阅:");
            ConsumerRecords<Integer, String> poll = consumer.poll(1000);
            poll.forEach(p->log.info(p.key()+":"+p.value()));
            consumer.commitSync();
        }
    }
}
